package com.group1.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV1;
import com.group1.realtime.bean.BroadCastAccTimeBean;
import com.group1.realtime.bean.BroadCastPersonNumBean;
import com.group1.realtime.bean.BroadCast_AvgPersonViewTimeBean;
import com.group1.realtime.common.Constant;
import com.group1.realtime.function.DimAsyncFunction;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.DimUtil;
import com.group1.realtime.util.DruidDSUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws_BroadCast_AvgPersonViewTime extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_BroadCast_AvgPersonViewTime().init(
                3233,
                2,
                "Dws_BroadCast_AvgPersonViewTime1",
                Constant.TOPIC_DWD_TRAFFIC_APPVIDEO
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {

        SingleOutputStreamOperator<BroadCast_AvgPersonViewTimeBean> beanStream = stream
                .map(JSON::parseObject)
                .map(value -> {

                    JSONObject data = value.getJSONObject("appVideo");

                    return BroadCast_AvgPersonViewTimeBean.builder()
                            .video_id(data.getString("video_id"))
                            .acc_time(data.getLong("play_sec"))
                            .person_num(1L)
                            .personViewTime(0.0)
                            .uid(value.getJSONObject("common").getString("uid"))
                            .ts(value.getLong("ts"))
                            .build();
                });

        /*SingleOutputStreamOperator<BroadCast_AvgPersonViewTimeBean> streamWithDim = AsyncDataStream.unorderedWait(
                beanStream,
                new DimAsyncFunction<BroadCast_AvgPersonViewTimeBean>() {

                    @Override
                    public String getTable() {
                        return "dim_video_info";
                    }

                    @Override
                    public String getId(BroadCast_AvgPersonViewTimeBean input) {
                        return input.getVideo_id();
                    }

                    @Override
                    public void addDim(BroadCast_AvgPersonViewTimeBean input,
                                       JSONObject dim) {
                        input.setChapter_id(dim.getString("CHAPTER_ID"));
                    }
                },
                60,   // 异步处理超时时间
                TimeUnit.SECONDS  // 超时赶时间的单位
        );*/
        SingleOutputStreamOperator<BroadCast_AvgPersonViewTimeBean> streamWithDim = beanStream
                .map(new RichMapFunction<BroadCast_AvgPersonViewTimeBean, BroadCast_AvgPersonViewTimeBean>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 1. 获取Phoenix连接
                        DruidDataSource dataSource = DruidDSUtil.createDataSource();
                        conn = dataSource.getConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        // 关闭Phoenix连接
                        conn.close();
                    }

                    @Override
                    public BroadCast_AvgPersonViewTimeBean map(BroadCast_AvgPersonViewTimeBean bean) throws Exception {
                        // 1. 查询sku_info
                        JSONObject skuInfo = DimUtil.readDimFromPhoenix(conn, "dim_video_info", bean.getVideo_id());
                        bean.setChapter_id(skuInfo.getString("CHAPTER_ID"));

                        return bean;
                    }
                });

        streamWithDim
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<BroadCast_AvgPersonViewTimeBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                                .withIdleness(Duration.ofSeconds(5))
                )
                .keyBy(BroadCast_AvgPersonViewTimeBean::getChapter_id)
                .reduce(new ReduceFunction<BroadCast_AvgPersonViewTimeBean>() {
                    @Override
                    public BroadCast_AvgPersonViewTimeBean reduce(BroadCast_AvgPersonViewTimeBean value1,
                                                                  BroadCast_AvgPersonViewTimeBean value2) throws Exception {
                        value1.setAcc_time(value1.getAcc_time() + value2.getAcc_time());
                        return value1;
                    }
                })
                .keyBy(BroadCast_AvgPersonViewTimeBean::getUid)
                .process(new KeyedProcessFunction<String, BroadCast_AvgPersonViewTimeBean, BroadCast_AvgPersonViewTimeBean>() {

                    private ValueState<BroadCast_AvgPersonViewTimeBean> singleUidState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        singleUidState = getRuntimeContext().getState(new ValueStateDescriptor<BroadCast_AvgPersonViewTimeBean>("singleUidState", BroadCast_AvgPersonViewTimeBean.class));
                    }

                    @Override
                    public void processElement(BroadCast_AvgPersonViewTimeBean value,
                                               Context ctx,
                                               Collector<BroadCast_AvgPersonViewTimeBean> out) throws Exception {

                        //第一条数据进来的时候, 状态应该是空, 更新状态
                        if (singleUidState.value() == null) {
                            // 1. 更新状态
                            singleUidState.update(value);

                            out.collect(singleUidState.value());
                        }

                    }
                })
                .keyBy(BroadCast_AvgPersonViewTimeBean::getChapter_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<BroadCast_AvgPersonViewTimeBean>() {
                            @Override
                            public BroadCast_AvgPersonViewTimeBean reduce(BroadCast_AvgPersonViewTimeBean value1,
                                                                          BroadCast_AvgPersonViewTimeBean value2) throws Exception {
                                value1.setPerson_num(value1.getPerson_num() + value2.getPerson_num());
                                //value1.setPersonViewTime((double) (value1.getAcc_time() + value2.getAcc_time() / value1.getPerson_num() + value2.getPerson_num()));
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<BroadCast_AvgPersonViewTimeBean, BroadCast_AvgPersonViewTimeBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<BroadCast_AvgPersonViewTimeBean> elements,
                                                Collector<BroadCast_AvgPersonViewTimeBean> out) throws Exception {

                                BroadCast_AvgPersonViewTimeBean bean = elements.iterator().next();

                                bean.setPersonViewTime((double) (bean.getAcc_time() / bean.getPerson_num()));

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setTs(ctx.currentProcessingTime());

                                out.collect(bean);
                            }
                        }
                )
                //.print();
                .addSink(FlinkSinkUtil.getClickHouseSink("dws_broadcast_avg_person_view_time", BroadCast_AvgPersonViewTimeBean.class));
    }

}

